package com.zxy.springboot_goods.Listeners;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * @Auther: lenovo
 * @Date: 2021/11/3 11:28
 * @Description:
 */
@Component
public class KafkaListeners{

    /*@Autowired
    SendMsgUtil sendMsgUtil;*/
    @Autowired
    private RedisTemplate redisTemplate;

    //Acknowledgment 消费方 ack确认机制
    @KafkaListener(topics = "look",id = "ack",containerFactory = "ackContainerFactory")
    public void onMessage(ConsumerRecord<String,String> data, Acknowledgment ack) {

        String key = data.key();
        Boolean hasKey = redisTemplate.hasKey(key);
        if(hasKey){
            System.out.println("重复消费");
        }else{
            String value = data.value();
            System.err.println(value);

            /*if(value.equals("send")){
                sendMsgUtil.sendTextEmail("3147448631@qq.com","周小勇","当前不是重复消费");
            }*/
            //手动提交offset
            ack.acknowledge();
        }

    }
}
